package Example

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

object Explore {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("Data Explore")
      //设置Hive连接参数
      .config("hive.metastore.uris", "thrift://master:9083")
      .config("spark.sql.warehouse.dir", "hdfs://master:8020/user/hive/warehouse")
      .enableHiveSupport()
      .getOrCreate()

    val media_index = spark.read.table("portrait.media_index")
    val billevent = spark.read.table("portrait.billevent")
    val orderData = spark.read.table("portrait.orderData")
    val usermsg = spark.read.table("portrait.usermsg")
    val userevent = spark.read.table("portrait.userevent")

    //数据总体概述，统计各表记录数与用户数（常规分析)  =>存入Mysql，存入txt吧，这个真没必要。
    println("用户信息表数量：" + usermsg.count())
    println("用户状态改变信息表数量：" + userevent.count())
    println("账单信息表数量：" + billevent.count())
    println("订单信息表数量：" + orderData.count())
    println("收视信息表数量：" + media_index.count())
    println("用户信息表涉及用户数：" + usermsg.select("phone_no").distinct().count())

    //统计用户收视记录观看时长的均值、最值和标准差(业务内容分析） =》duration以毫秒为单位。这里是转化成分钟查看。
    //describe()相当于R语言的summary咯。对数据做一个简单的分析
    media_index.selectExpr("duration/(1000*60) as duration_minute").describe().show()

    //统计用户月均观看时长（业务内容分析）
    println("收视记录时间统计：\n")
    media_index.select("origin_time").describe().show() //共涵盖三个月的数据  对起始时间？
    // 这单位有问题啊。单看数据duration列的单位应该是毫秒才对啊。这怎么来的(3*1000*60*60)？
    val perUserDuration = media_index
      .groupBy("phone_no").agg((sum("duration") / (3 * 1000 * 60 * 60))
      .alias("duration_avg"))
    perUserDuration.orderBy(desc("duration_avg")).show(5) //查看月均观看时长前5的用户

    //统计重复数据的用户（异常数据分析）
    println("用户信息表重复记录统计：")
    usermsg.groupBy("phone_no").count().filter("count>1").count()
    usermsg.groupBy("phone_no").count().orderBy(desc("count")).show()

    //特殊线路的用户（异常数据分析）
    //统计各表的owner_code类别数
    val usermsgCode = usermsg.groupBy("owner_code").count()
    val usereventCode = userevent.groupBy("owner_code").count()
    val billeventsCode = billevent.groupBy("owner_code").count()
    val orderCode = orderData.groupBy("owner_code").count()
    val mediaCode = media_index.groupBy("owner_code").count()

    println("用户信息表特殊线路用户分布情况：")
    usermsgCode.show()
    println("用户状态变更信息表特殊线路用户分布情况：")
    usereventCode.show()
    println("账单表特殊线路用户分布情况：")
    billeventsCode.show()
    println("订单表特殊线路用户分布情况：")
    orderCode.show()
    println("收视信息表特殊线路用户分布情况：")
    mediaCode.show()

    //政企用户（异常数据分析）
    //统计各表的owner_name类别数
    val usermsgOwnerName = usermsg.groupBy("owner_name").count()
    val usereventOwnerName = userevent.groupBy("owner_name").count()
    val billeventsOwnerName = billevent.groupBy("owner_name").count()
    val orderOwnerName = orderData.groupBy("owner_name").count()
    val mediaOwnerName = media_index.groupBy("owner_name").count()

    println("用户信息表政企用户分布情况：")
    usermsgOwnerName.show()
    println("用户状态变更信息表政企用户分布情况：")
    usereventOwnerName.show()
    println("账单表政企用户分布情况：")
    billeventsOwnerName.show()
    println("订单表政企用户分布情况：")
    mediaOwnerName.show()
    println("收视信息表政企用户分布情况：")
    orderOwnerName.show()

    //分析目标业务数据
    val nums = usermsg.count()

    //统计用户基本信息中sm_name各类别的数量和占比以及run_name各类别情况
    println("用户信息表sm_name分布：")
    usermsg.groupBy("sm_name").count().withColumn("percent", col("count") / nums).show()
    println("用户信息表run_name分布：")
    usermsg.groupBy("run_name").count().show()

    //分析用户收视行为无效数据（观看时长过长与过短)
    val total = media_index.count() //统计总记录数
    //以小时为单位分析用户收视行为
    val mediaHours = media_index.withColumn("hours", floor(col("duration") / (1000 * 60 * 60))) //增加一个字段hours：将观看时长转为小时并向下取整
    val hoursNum = mediaHours.groupBy("hours").count().withColumn("percent", col("count") / total) //统计各hours值的记录数和各类所占的比例
    println("以小时为单位，用户收视情况：")
    hoursNum.show()

    //以分钟为单位分析用户收视行为
    val oneHour = mediaHours.where("hours<1") //统计收视行为中观看时长小于1小时的各区间记录数
    val mediaMinutes = oneHour.withColumn("minutes", floor(col("duration") / (1000 * 60))) //增加一个字段minutes
    val minutesNum = mediaMinutes.groupBy("minutes").count()
      .withColumn("percent", col("count") / total) //统计各minutes值的记录数
    println("以分钟为单位，用户收视情况：")
    minutesNum.orderBy(asc("minutes")).show(10) //按字段minutes升序排序并显示前10条结果

    //以秒为单位分析用户收视行为
    val oneMinute = mediaMinutes.where("minutes<1") //统计用户收视行为中小于1分钟各区间的记录数
    val mediaSeconds = oneMinute.withColumn("seconds", floor(col("duration") / 1000)) //新增字段seconds：将观看时长转为以每秒为间隔的数并向下取整
    val secondsNum = mediaSeconds.groupBy("seconds").count()
      .withColumn("percent", col("count") / total) //统计各seconds值的记录数
    println("以秒为单位，用户收视情况：")
    secondsNum.orderBy(asc("seconds")).show(10) //按字段seconds升序排序并显示前10条结果

    //统计用户收视行为无效数据(res_type=0且起始结束为00秒)
    val invalidData = media_index.filter("res_type=0")
      .filter(col("origin_time").endsWith("00") && col("end_time").endsWith("00"))
    println("无效数据条数：" + invalidData.count())
    println("无效数据：")
    invalidData.show()

    //消费水平标签阈值探索
    //统计电视用户月均消费情况
    val tvBilleventsData = billevent.filter("sm_name like '%电视%' and sm_name !='模拟有线电视'") //筛选电视用户
    val avgTVBillData = tvBilleventsData.groupBy("phone_no").agg((sum(col("should_pay") - col("favour_fee")) / 7).alias("avg_fee")) //统计1-7月平均每月的每个用户的消费金额
    println("电视用户月均消费金额统计：")
    avgTVBillData.select("avg_fee").describe().show() //描述性统计
    val rangeTVBillData = avgTVBillData.withColumn("range_fee", floor(col("avg_fee") / 10) * 10) //每月消费金额按10的间隔划分
    val rangeTVCount = rangeTVBillData.groupBy("range_fee").count() //按ragne_fee分组统计记录数
    println("电视用户月均消费排名情况：")
    rangeTVCount.orderBy(asc("range_fee")).show(30) //按range_fee顺序显示前30结果
    //查看过滤后的电视用户月均消费金额及标准差
    println("排除消费过高过低记录后其余电视用户的月均消费金额及标准差：")
    avgTVBillData.filter("avg_fee>=-10 and avg_fee<=90")
      .select(avg("avg_fee"), stddev("avg_fee")).show()

    //统计宽带用户月均消费情况
    val netBilleventsData = billevent.filter("sm_name like '%珠江宽频%'") //筛选宽带用户
    val avgNetBillData = netBilleventsData.groupBy("phone_no")
      .agg((sum(col("should_pay") - col("favour_fee")) / 7).alias("avg_fee")) //统计1-7月平均每月的每个用户的消费金额
    println("宽带用户月均消费金额统计：")
    avgNetBillData.summary().show() //描述性统计
    val rangeNetBillData = avgNetBillData.withColumn("range_fee", floor(col("avg_fee") / 10) * 10) //每月消费金额按10的间隔划分
    val rangeNetCount = rangeNetBillData.groupBy("range_fee").count() //按ragne_fee分组统计记录数
    println("宽带用户月均消费排名情况：")
    rangeNetCount.orderBy(asc("range_fee")).show(30) //按range_fee顺序显示前30结果

    //统计分析电视用户入网时长
    val tvUsermsgData = usermsg.filter("sm_name like '%电视%' and sm_name !='模拟有线电视'") //筛选电视用户
    val tvFilteredUsermsgData = tvUsermsgData.filter("open_time != 'NULL'") //过滤open_time为空值的记录
    //把用户开户时间与当前时间作差并向下取整（单位为年）
    val yearTVUserData = tvFilteredUsermsgData
      .groupBy("phone_no").agg(max(floor(datediff(current_date(), col("open_time")) / 365))
      .alias("years"))
    println("查看电视用户入网时长的分位点")
    yearTVUserData.select("years").summary().show()
    val yearTVUserCount = yearTVUserData.groupBy("years").count() //分组统计每个入网户时长值的用户数
    println("电视用户入网时长分布情况：")
    yearTVUserCount.orderBy(asc("years")).show(30)

    //统计分析宽带用户入网时长
    val netUsermsgData = usermsg.filter("sm_name='珠江宽频'") // 筛选宽带用户
    val netFilteredUsermsgData = netUsermsgData.filter("open_time != 'NULL'") //过滤open_time为空值的记录
    //把用户开户时间与当前时间作差并向下取整（单位为年）
    val yearNetUserData = netFilteredUsermsgData
      .groupBy("phone_no").agg(max(floor(datediff(current_date(), col("open_time")) / 365))
      .alias("years"))
    println("查看宽带用户入网时长的分位点")
    yearNetUserData.select("years").summary().show()
    val yearNetUserCount = yearNetUserData.groupBy("years").count() //分组统计每个入网时长值的用户数
    println("电视用户入网时长分布情况：")
    yearNetUserCount.orderBy(asc("years")).show()
  }
}
